package cn.gwm.flink.streaming.function.apply;

import cn.gwm.flink.streaming.constant.BaseFields;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.*;

/**
 * @ClassName OdsCoProcessFunction
 * @Author xzh
 * @Date 2022/12/27 13:42
 **/
@Slf4j
public class OdsCoProcessFunction extends CoProcessFunction<JSONObject,JSONObject,JSONObject> {
    /*
     * 定义状态变量，用来保存已经到达的事件
     **/
    private transient ValueState<JSONObject> canObj;
    private transient ValueState<Long> onTimerTs;
    private transient MapState<String, JSONObject> mapState;
    private final long DELAY_TIME = 60*60*1000L;

    @Override
    public void open(Configuration parameters) throws Exception {
        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.seconds(300))
                        .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                        .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired).build();
        canObj = getRuntimeContext().getState(new ValueStateDescriptor<>("canObj",JSONObject.class));
        MapStateDescriptor<String, JSONObject> mapState = new MapStateDescriptor<>("mapState", String.class, JSONObject.class);
        mapState.enableTimeToLive(ttlConfig);
        this.mapState = getRuntimeContext().getMapState(mapState);
        onTimerTs = getRuntimeContext().getState(new ValueStateDescriptor<>("onTimerTs", Long.class));
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
        clearObj(out,ctx);
    }

    private void clearObj(Collector<JSONObject> out,Context context) throws Exception {
        if(canObj!=null&&canObj.value()!=null){
            JSONObject object = canObj.value();
            if(object!=null){
                if(!mapState.isEmpty()){
                    if(mapState.contains(object.getStr(BaseFields.tid))){
                        JSONObject obj = mapState.get(object.getStr(BaseFields.tid));
                        if(obj!=null) {
                            object.set("longitude", obj.getStr("longitude"));
                            object.set("latitude", obj.getStr("latitude"));
                            object.set(BaseFields.MATCH_ERROR, 0);
                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+1000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+1000L));
                        if(obj!=null) {
                            object.set("longitude", obj.getStr("longitude"));
                            object.set("latitude", obj.getStr("latitude"));
                            object.set(BaseFields.MATCH_ERROR, 1);
                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-1000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-1000L));
                        if(obj!=null) {
                            object.set("longitude", obj.getStr("longitude"));
                            object.set("latitude", obj.getStr("latitude"));
                            object.set(BaseFields.MATCH_ERROR, -1);
                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+2000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+2000L));
                        if(obj!=null) {
                            object.set("longitude", obj.getStr("longitude"));
                            object.set("latitude", obj.getStr("latitude"));
                            object.set(BaseFields.MATCH_ERROR, 2);
                        }
                    }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-2000L))){
                        JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-2000L));
                        if(obj!=null) {
                            object.set("longitude", obj.getStr("longitude"));
                            object.set("latitude", obj.getStr("latitude"));
                            object.set(BaseFields.MATCH_ERROR, -2);
                        }
                    }
                    Iterator<Map.Entry<String, JSONObject>> iterator = mapState.iterator();
                    while (iterator.hasNext()){
                        Map.Entry<String, JSONObject> next = iterator.next();
                        //当can数据延迟落后gps10分钟后，放弃合并
                        if( next!=null && (next.getKey().compareTo(object.getStr(BaseFields.tid))<=0
                                || next.getKey().compareTo(String.valueOf(object.getLong(BaseFields.tid)+DELAY_TIME)) >0)){
                            iterator.remove();
                        }
                    }
                }
                out.collect(object);
                canObj.clear();
                onTimerTs.clear();
            }
        }
    }

    @Override
    public void processElement1(JSONObject object, Context context, Collector<JSONObject> collector) throws Exception {
        if(canObj.value()!=null&&onTimerTs.value()!=null){
            context.timerService().deleteProcessingTimeTimer(onTimerTs.value());
            clearObj(collector,context);
        }
        if(!mapState.isEmpty()){
            if(mapState.contains(object.getStr(BaseFields.tid))){
                JSONObject obj = mapState.get(object.getStr(BaseFields.tid));
                if(obj!=null){
                    object.set("longitude", obj.getStr("longitude"));
                    object.set("latitude", obj.getStr("latitude"));
                    object.set(BaseFields.MATCH_ERROR, 0);
                    collector.collect(object);
                }else{
                    canObj.update(object);
                    long ts = context.timerService().currentProcessingTime() + 8000L;
                    onTimerTs.update(ts);
                    context.timerService().registerProcessingTimeTimer(ts);

                }
            }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+1000L))){
                JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+1000L));
                if(obj!=null){
                    object.set("longitude", obj.getStr("longitude"));
                    object.set("latitude", obj.getStr("latitude"));
                    object.set(BaseFields.MATCH_ERROR, 1);
                    collector.collect(object);
                }else{
                    canObj.update(object);
                    long ts = context.timerService().currentProcessingTime() + 8000L;
                    onTimerTs.update(ts);
                    context.timerService().registerProcessingTimeTimer(ts);

                }
            }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-1000L))){
                JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-1000L));
                if(obj!=null){
                    object.set("longitude", obj.getStr("longitude"));
                    object.set("latitude", obj.getStr("latitude"));
                    object.set(BaseFields.MATCH_ERROR, -1);
                    collector.collect(object);
                }else{
                    canObj.update(object);
                    long ts = context.timerService().currentProcessingTime() + 8000L;
                    onTimerTs.update(ts);
                    context.timerService().registerProcessingTimeTimer(ts);

                }
            }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)+2000L))){
                JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)+2000L));
                if(obj!=null){
                    object.set("longitude", obj.getStr("longitude"));
                    object.set("latitude", obj.getStr("latitude"));
                    object.set(BaseFields.MATCH_ERROR, 2);
                    collector.collect(object);
                }else{
                    canObj.update(object);
                    long ts = context.timerService().currentProcessingTime() + 8000L;
                    onTimerTs.update(ts);
                    context.timerService().registerProcessingTimeTimer(ts);
                }
            }else if(mapState.contains(String.valueOf(object.getLong(BaseFields.tid)-2000L))){
                JSONObject obj = mapState.get(String.valueOf(object.getLong(BaseFields.tid)-2000L));
                if(obj!=null){
                    object.set("longitude", obj.getStr("longitude"));
                    object.set("latitude", obj.getStr("latitude"));
                    object.set(BaseFields.MATCH_ERROR, -2);
                    collector.collect(object);
                }else{
                    canObj.update(object);
                    long ts = context.timerService().currentProcessingTime() + 8000L;
                    onTimerTs.update(ts);
                    context.timerService().registerProcessingTimeTimer(ts);

                }
            }else{
                canObj.update(object);
                long ts = context.timerService().currentProcessingTime() + 8000L;
                onTimerTs.update(ts);
                context.timerService().registerProcessingTimeTimer(ts);
            }
            Iterator<Map.Entry<String, JSONObject>> iterator = mapState.iterator();
            while (iterator.hasNext()){
                Map.Entry<String, JSONObject> next = iterator.next();
                if( next!=null && (next.getKey().compareTo(object.getStr(BaseFields.tid))<=0
                        || next.getKey().compareTo(String.valueOf(object.getLong(BaseFields.tid)+DELAY_TIME))>0)){
                    iterator.remove();
                }
            }

        }else{
            canObj.update(object);
            long ts = context.timerService().currentProcessingTime() + 8000L;
            onTimerTs.update(ts);
            context.timerService().registerProcessingTimeTimer(ts);
//            context.timerService().registerEventTimeTimer(object.getLong(BaseFields.tid) + 2000L);
        }
    }

    @Override
    public void processElement2(JSONObject obj, Context context, Collector<JSONObject> collector) throws Exception {
        if(canObj!=null&&canObj.value()!=null){
            JSONObject object = canObj.value();
            if(object!=null){
                if(object.getStr(BaseFields.tid).equals(obj.getStr(BaseFields.tid))){
                    object.set("longitude", obj.getStr("tboxMoniLongitude"));
                    object.set("latitude", obj.getStr("tboxMoniLatitude"));
                    object.set(BaseFields.MATCH_ERROR, 0);
                    collector.collect(object);
                    canObj.clear();
                    onTimerTs.clear();
                }else {
                    putGpsObj(obj);
                }
            }else{
                putGpsObj(obj);
            }
        }else{
            putGpsObj(obj);
        }

    }

    private void putGpsObj(JSONObject obj) throws Exception {
        JSONObject object = new JSONObject();
        object.set("longitude", obj.getStr("tboxMoniLongitude"));
        object.set("latitude", obj.getStr("tboxMoniLatitude"));
        object.set(BaseFields.tid, obj.getStr(BaseFields.tid));
        object.set(BaseFields.vin, obj.getStr(BaseFields.vin));
        mapState.put(object.getStr(BaseFields.tid), object);
    }
}
